package com.abyss.transformation;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FilterOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.operators.PartitionOperator;
import org.apache.flink.api.java.tuple.Tuple2;
/**
 * @author Abyss
 * @date 2020/10/4
 * @description
 */
public class RebalanceDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. Source 1~100 的序列
        DataSource<Long> source = env.generateSequence(1, 100);
        FilterOperator<Long> filter = source.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value > 8;
            }
        });

        PartitionOperator<Long> rebalance = filter.rebalance();
        MapOperator<Long, Tuple2<Integer, Integer>> map = rebalance.map(new RichMapFunction<Long, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(Long value) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(), 1);
            }
        });

        map.groupBy(0).sum(1).print();
    }
}
